package pbc.gx.SaveJson.kafka.kafka2redis;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class Kafka2Redis {
    private static final ExecutorService executorRedis = Executors.newFixedThreadPool(10);
    private static final ExecutorService executorMysql = Executors.newFixedThreadPool(10);

    private final JdbcTemplate jdbcTemplate;  //这个是系统自带的
    private final RedisTemplate redisTemplate;

    public Kafka2Redis(JdbcTemplate jdbcTemplate, RedisTemplate redisTemplate) {
        this.jdbcTemplate = jdbcTemplate;
        this.redisTemplate = redisTemplate;
    }

    @KafkaListener(topics = "jsonTopic", groupId = "jsonGoup")
    public void Kafka2RedisMysql(List<ConsumerRecord<String, JSONObject>> records) {
        if (records == null || records.size() <= 0)
            return;
        executorRedis.execute(new Save2RedisThread(redisTemplate, records));
        executorMysql.execute(new Save2MysqlThread(jdbcTemplate, records));
    }
}
